package com.sia.springcloud.process.consumer;

import com.alibaba.fastjson.JSONObject;
import com.sia.springcloud.process.influxDB.InfluxDbTool;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.influxdb.dto.BatchPoints;
import org.influxdb.dto.Point;

import java.util.*;
import java.util.concurrent.TimeUnit;

/**
 * @Author: Sirius
 * @Date: 2022/5/14 18:02
 * @Description:
 **/
@Slf4j
public class KafkaConsumerAbb implements Runnable {


    // 创建一个Kafka消费者
    private KafkaConsumer<String, String> consumer;

    //一个容器，其中包含特定主题的每个分区的ConsumerRecord列表。
    // 对于Consumer.poll（long）操作返回的每个主题分区，都有一个ConsumerRecord列表。
    private ConsumerRecords<String, String> msgList;
    private InfluxDbTool influxDbTool;
    private  String topic;
    private static final String GROUPID = "etl";



    public KafkaConsumerAbb(String topicName,InfluxDbTool influxDbTool) {
        this.topic = topicName;
        this.influxDbTool=influxDbTool;
        init();
    }

    @Override
    public void run() {
        Map<Integer,Map<Integer,String>> cache = new HashMap<>();

        System.out.println("---------开始消费---------");
        List<String> list=new ArrayList<String>();
        List<Long> list2=new ArrayList<Long>();

        // 构造一个批量数据的点
        BatchPoints batchPoints  = BatchPoints.builder().build();

        Map<String,String> tempMap=new HashMap<String,String>();

        // 循环执行
        while (true) {
            try {
                msgList = consumer.poll(100);
                if(null!=msgList){


                    for (ConsumerRecord<String, String> record : msgList) {
                        list.add(record.value());
                        list2.add(record.offset());
                        if(record.topic().equals("abb") && null != record && record.value().contains("{")){
                            System.out.println("主题:"+ record.topic()+"内容: "+record.value()+"offset: "+record.offset());
                            JSONObject jsonObject=JSONObject.parseObject(record.value());

                            /**
                             * 设备名作为表名，字段名作为tag
                             */
                            Point point = Point.measurement(jsonObject.getString("deviceName"))
                                    .addField("collectTime",jsonObject.getString("time"))
                                    .tag("variableValue",jsonObject.getString("value"))
                                    .tag("variableName",jsonObject.getString("abbVariablename"))
                                    .tag("deviceName",jsonObject.getString("deviceName"))
                                    .time(System.currentTimeMillis(), TimeUnit.MILLISECONDS).build();
                            batchPoints.point(point);//将point放入batch中
                        }
                    }
                    // 将打包好的记录保存到influxDB中，数据库名为“abb”
                    influxDbTool.saveInfluxDb(batchPoints,"abb");

                    // 业务处理成功后，手动提交
                    consumer.commitSync();

                    // 清空三个集合
                    batchPoints.getPoints().clear();
                    list.clear();
                    list2.clear();

                }else{
                    // 如果拉取的集合为空，线程休眠1秒？为什么？
                    Thread.sleep(1000);
                }
            }catch (Exception e){
                //logger.error("abb存入influxdb失败:"+e.getMessage());
            }
        }
    }

    private void init() {
        Properties props = new Properties();

        //kafka消费的的地址
//        props.put("bootstrap.servers", "172.20.3.81:9092");
        props.put("bootstrap.servers", "127.0.0.1:9092");
        //组名 不同组名可以重复消费
        props.put("group.id", GROUPID);
        //是否自动提交
        props.put("enable.auto.commit", "false");
        //超时时间
        props.put("session.timeout.ms", "30000");
        //一次最大拉取的条数
        props.put("max.poll.records", 5000);
//		earliest当各分区下有已提交的offset时，从提交的offset开始消费；无提交的offset时，从头开始消费
//		latest
//		当各分区下有已提交的offset时，从提交的offset开始消费；无提交的offset时，消费新产生的该分区下的数据
//		none
//		topic各分区都存在已提交的offset时，从offset后开始消费；只要有一个分区不存在已提交的offset，则抛出异常
        props.put("auto.offset.reset", "earliest");
        //序列化
        props.put("key.deserializer", StringDeserializer.class.getName());
        props.put("value.deserializer", StringDeserializer.class.getName());
        this.consumer = new KafkaConsumer<String, String>(props);
        //订阅主题列表topic
        this.consumer.subscribe(Arrays.asList(topic));

        System.out.println("abb初始化!");
    }
}

